/*
 * FileName: KafKaProducerUtil.java
 * Author:   zzw
 * Date:     2018年05月24日
 * Description:
 */
package com.chezhibao.trace.util;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Map;
import java.util.Properties;

/**
 * 〈Kafka工具类〉<br>
 * 〈Kafka工具类〉
 *
 * @author zzw
 * @see [相关类/方法]（可选）
 * @since [产品/模块版本]（可选）
 */
public class KafKaProducerUtil {

    private static String KAFKA_METADATA_BROKER_LIST="hadoop1:9092";
    private static String KAFKA_KEY_SERIALIZER_CLASS="kafka.serializer.StringEncoder";
    private static String KAFKA_SERIALIZER_CLASS="kafka.serializer.StringEncoder";
    private static String KAFKA_REQUEST_REQUIRED_ACKS="1";
    private static String KAFKA_COMPRESSION_CODEC="snappy";
    private static String KAFKA_MESSAGE_SEND_MAX_RETRIES="2";
    private static String KAFKA_RECONNECT_BACKOFF_MS="100";
    private static String KAFKA_REQUEST_TIMEOUT_MS="10000";
    private static String KAFKA_PRODUCER_TYPE="sync"; //同步处理
    private static String KAFKA_QUEUE_BUFFERING_MAX_MS=""; //消息缓存最长时间 异步方式下配置
    private static String KAFKA_QUEUE_BUFFERING_MAX_MESSAGES="";//缓存消息的最大值
    private static String KAFKA_QUEUE_ENQUEUE_TIMEOUT_MS="";//消息进入队列的的等待时间
    private static String KAFKA_BATCH_NUM_MESSAGES="";//批量消息条数
    private static String KAFKA_SEND_BUFFER_BYTES="";//socket缓存大小

    private static Producer<String, String> producer;

    private static Producer<String, String> getProducer()
    {

        if(producer==null) {
            synchronized (KafKaProducerUtil.class) {
                Properties pro=new Properties();
                pro.setProperty("metadata.broker.list",KAFKA_METADATA_BROKER_LIST);
                pro.setProperty("request.required.acks", KAFKA_REQUEST_REQUIRED_ACKS);
                pro.setProperty("producer.type", KAFKA_PRODUCER_TYPE);
                pro.setProperty("serializer.class", KAFKA_SERIALIZER_CLASS);
                pro.setProperty("request.timeout.ms", KAFKA_REQUEST_TIMEOUT_MS);
                pro.setProperty("key.serializer.class", KAFKA_KEY_SERIALIZER_CLASS);
                pro.setProperty("compression.codec", KAFKA_COMPRESSION_CODEC);
                pro.setProperty("message.send.max.retries", KAFKA_MESSAGE_SEND_MAX_RETRIES);
                pro.setProperty("retry.backoff.ms", KAFKA_RECONNECT_BACKOFF_MS);
                producer = new KafkaProducer<String, String>(pro);
            }
        }
        return producer;
    }
}
